package cn.linjianhui.demo.flink;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;

import java.util.Random;
import java.util.UUID;

public class JDBCTableSinkTest {
    private static final TypeInformation[] FIELD_TYPES = new TypeInformation[]{
            BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO
    };

    public static void main(String[] args) throws Exception {
        JDBCAppendTableSink sink = JDBCAppendTableSink.builder()
                .setDrivername("com.mysql.cj.jdbc.Driver")
                .setDBUrl("jdbc:mysql://127.0.0.1:3306/sys?characterEncoding=utf-8&serverTimezone=GMT")
                .setUsername("root")
                .setPassword("qwe321")
                .setQuery("INSERT INTO rand VALUES (?, ?) ON DUPLICATE KEY UPDATE id = values(id), rand = values(rand)")
                .setParameterTypes(FIELD_TYPES)
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Row> ds = env.fromElements(Row.of(UUID.randomUUID().toString(), new Random().nextInt(10000)),
                Row.of(UUID.randomUUID().toString(), new Random().nextInt(10000)));
        sink.emitDataStream(ds);

        env.execute("JDBCTableSinkTest");
    }
}
